package io.openmessaging.storage;

import io.openmessaging.Conf;
import io.openmessaging.Message;
import io.openmessaging.Topic;
import io.openmessaging.io.FileChannelWrapper;
import io.openmessaging.map.TopicQueueOffsetMap;
import io.openmessaging.utils.Statistic;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author cuijd    2021/9/26 9:11
 */
public class SSDStorage implements Storage {
    private int writeThreadSize = 4;//落盘线程数
    private FileChannelWrapper[] fileChannels = new FileChannelWrapper[writeThreadSize];
    private final LinkedBlockingQueue<Message> bufferQueue = new LinkedBlockingQueue<>();
    private TopicQueueOffsetMap<FileIndex> indices = new TopicQueueOffsetMap<>();

    public SSDStorage() {
        for (byte i = 0; i < writeThreadSize; i++) {
            fileChannels[i] = new FileChannelWrapper(Conf.ssdPath + i);
            if (fileChannels[i].size() == 0)
                continue;
            //根据已有数据恢复索引
            long readPos = 0;
            ByteBuffer head = ByteBuffer.allocate(9);//消息头，储存 topicId(1) queueId(2) offset(4) dataLength(2) 共9字节
            while (readPos < fileChannels[i].size()) {
                head.clear();
                readPos += fileChannels[i].readAndFlip(head, readPos);
                FileIndex index = indices.set(head.get(), head.getShort(), head.getInt(), new FileIndex(i, readPos - 9, head.getShort()));
                readPos += index.length;
            }
        }

        for (byte i = 0; i < writeThreadSize; i++) {
            new BlockingQueueWriter(fileChannels[i], i).start();
        }
    }

    @Override
    public void save(Message m) {
        try {
            synchronized (m) {
                bufferQueue.put(m);
                m.wait();//等待force之后返回
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    private final class BlockingQueueWriter extends Thread {
        FileChannelWrapper channel;
        byte fileId;

        public BlockingQueueWriter(FileChannelWrapper channel, byte fileId) {
            this.channel = channel;
            this.fileId = fileId;
            setName("SSD_W" + fileId);
        }

        @Override
        public void run() {
            try {
                while (true) {
                    while (true) {
                        Message m = bufferQueue.take();
                        List<Message> messages = new ArrayList<>(bufferQueue.size() + 1);
                        for (; m != null; m = bufferQueue.poll()) {
                            messages.add(m);
                        }
                        writeAndNotify(messages);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        ByteBuffer head = ByteBuffer.allocateDirect(9);//消息头，储存 topicId(1) queueId(2) offset(4) dataLength(2) 共9字节

        private void writeAndNotify(List<Message> messages) {
            for (Message m : messages) {
                head.clear();
                head.put(m.topic.id).putShort((short) m.queueId).putInt((int) m.offset).putShort((short) m.data.remaining()).flip();
                //构造消息的文件索引，等append线程被唤醒后再使索引生效
                FileIndex index = new FileIndex(fileId, channel.size(), (short) m.data.remaining());//顺序不能变，记录当前position
                indices.set(m.topic.id, m.queueId, m.offset, index);
//                Log.i("SSDStorage.writeAndNotify topicId=", m.topic.id, m, index);
                channel.write(head, m.data);
            }

            channel.force();
            for (Message m : messages) {
                synchronized (m) {
                    m.notify();
                }
            }
            Statistic.SSD.force(messages.size());
        }
    }

    @Override
    public ByteBuffer get(Topic topic, int queueId, long offset) {
        return get(topic, queueId, offset, null);
    }

    /**
     * 先判断有没有落盘，如果没有，返回null，否则先从缓存中取，如果缓存取不到，再从 ssd 读
     */
    public ByteBuffer get(Topic topic, int queueId, long offset, Storage cache) {
        FileIndex index = this.indices.get(topic.id, queueId, offset);
        if (index == null)
            return null;

        ByteBuffer readBuffer = null;
        if (cache != null)
            readBuffer = cache.get(topic, queueId, offset);

        if (readBuffer == null) {
            readBuffer = ByteBuffer.allocate(index.length);
            fileChannels[index.fileId].readAndFlip(readBuffer, index.position + 9);//跳过header

            Statistic.SSD.read(readBuffer.remaining());
        }

        return readBuffer;
    }

}
